package com.shujia.dwi

import java.awt.geom.Point2D
import java.text.SimpleDateFormat

import com.shujia.grid.Grid
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

import scala.collection.mutable.ListBuffer

object StayPointApp extends Logging {

  def main(args: Array[String]): Unit = {


    /**
      * 停留表处理逻辑
      * 将同一个人在同一个网格中的多条数据合并成一条数据
      * 1、将同一个人一天内的所有数据分到一个组类
      * 2、安装时间对一个人的数据进行排序
      * 3、循环数据，一次判断是否在一个网格内
      *
      *
      */

    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id = args(0)

    log.info(s"当前时间分区为：$day_id")

    val spark: SparkSession = SparkSession.builder()
      .appName("StayPointApp")
      .config("spark.sql.shuffle.partitions", "40")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._
    /**
      *
      * mdn string comment '手机号码'
      * ,start_time string comment '业务时间'
      * ,county_id string comment '区县编码'
      * ,longi string comment '经度'
      * ,lati string comment '纬度'
      * ,bsid string comment '基站标识'
      * ,grid_id string comment '网格号'
      * ,biz_type string comment '业务类型'
      * ,event_type string comment '事件类型'
      * ,data_source string comment '数据源'
      *
      */

    //读取融合表
    val mergeLocation: DataFrame = spark.sql(s"select mdn,start_time,county_id,grid_id from dwi.dwi_res_regn_mergelocation_msk_d where day_id=$day_id")

    //将数据转换成kv格式，key使用手机号，value是时间，区县编号，网格编号
    val kvRDD: RDD[(String, (String, String, String, String))] = mergeLocation
      .rdd
      .map {
        case Row(mdn: String, start_time: String, county_id: String, grid_id: String) =>
          //取一个时间
          val endTime: String = start_time.split(",")(0)
          val startTime: String = start_time.split(",")(1)

          //以手机号作为key
          (mdn, (endTime, startTime, county_id, grid_id))
      }


    //按照手机号进行分组
    val groupRDD: RDD[(String, Iterable[(String, String, String, String)])] = kvRDD.groupByKey()


    /**
      *
      * 2、安装时间对一个人的数据进行排序
      * 3、循环数据，一次判断是否在一个网格内
      * 4、整理数据格式，获取网格中心点经纬度，计算停留时间
      */
    val pointRDD: RDD[(String, String, String, String, String, String, String, String)] = groupRDD.flatMap {
      case (mdn: String, points: Iterable[(String, String, String, String)]) =>


        //对一个一天所有的点按照时间进行升序排序
        val sortPoint: List[(String, String, String, String)] = points.toList.sortBy(_._1)


        //距离最终合并的结果
        val resultPoints = new ListBuffer[(String, String, String, String)]


        //第一个网格编号
        var headGrud: String = sortPoint.head._4
        //第一个点的时间
        var poinStartTime: String = sortPoint.head._2
        //结束时间
        var pointEndTime: String = sortPoint.head._1

        var last_county_id: String = sortPoint.head._3

        //遍历一个人所有的点
        sortPoint.foreach {
          case (endTime: String, startTime: String, county_id: String, grid_id: String) =>

            //如果当前网格和上一个网格是一样的，就是同一个组
            if (grid_id.equals(headGrud)) {
              headGrud = grid_id
              //更新结束时间
              pointEndTime = endTime

              last_county_id = county_id
            } else {

              //将上一个组保存起来
              resultPoints.+=((grid_id, poinStartTime, pointEndTime, county_id))

              //切换下一个组
              headGrud = grid_id
              poinStartTime = startTime
              pointEndTime = endTime
            }
        }


        //处理最后一组
        resultPoints.+=((headGrud, poinStartTime, pointEndTime, last_county_id))


        /**
          *
          * mdn string comment '用户手机号码'
          * ,longi string comment '网格中心点经度'
          * ,lati string comment '网格中心点纬度'
          * ,grid_id string comment '停留点所在电信内部网格号'
          * ,county_id string comment '停留点区县'
          * ,duration string comment '机主在停留点停留的时间长度（分钟）,lTime-eTime'
          * ,grid_first_time string comment '网格第一个记录位置点时间（秒级）'
          * ,grid_last_time string comment '网格最后一个记录位置点时间（秒级）'
          *
          */

        //整理数据


        //整个作为外面flatMap的返回值
        val resultGrid = resultPoints.map {
          case (grid_id: String, poinStartTime: String, pointEndTime: String, county_id: String) =>

            //获取网格中心点经纬度
            //Grid 是处理网格的工具
            val p: Point2D.Double = Grid.getCenter(grid_id.toLong)

            val longi: String = p.getX.formatted("%.4f")
            val lati: String = p.getY.formatted("%.4f")

            //计算停留时间
            val format = new SimpleDateFormat("yyyyMMddHHmmss")
            val startTs: Long = format.parse(poinStartTime).getTime
            val endTs: Long = format.parse(pointEndTime).getTime


            ///停留时间
            val duration: String = ((endTs - startTs) / 60000).toString

            (mdn, longi, lati, grid_id, county_id, duration, poinStartTime, pointEndTime)
        }


        //返回一个集合
        resultGrid

    }

    //保存数据
    pointRDD
      .toDF()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_staypoint_msk_d/day_id=$day_id")

    spark.sql(s"alter table dwi.dwi_staypoint_msk_d add if not exists partition(day_id='$day_id') ")


  }

}
